using System.Threading.Channels;
using System.Reactive.Subjects;
using System.Reactive.Linq;
using System.Reactive.Concurrency;
using Microsoft.Extensions.Logging;

namespace ReactiveChannelMessaging.Core;

/// <summary>
/// 高级反应式消息总线，结合Channel和Reactive Extensions
/// </summary>
public class ReactiveMessageBus : IAsyncDisposable
{
    private readonly Channel<Message> _channel;
    private readonly ChannelWriter<Message> _writer;
    private readonly ChannelReader<Message> _reader;
    private readonly Subject<Message> _messageSubject;
    private readonly ILogger<ReactiveMessageBus> _logger;
    private readonly CancellationTokenSource _cancellationTokenSource;
    private readonly Task _processingTask;

    // 按消息类型分发的Observable流
    private readonly IObservable<DataMessage> _dataMessages;
    private readonly IObservable<EventMessage> _eventMessages;
    private readonly IObservable<CommandMessage> _commandMessages;
    private readonly IObservable<ResponseMessage> _responseMessages;

    // 错误处理和重试机制
    private readonly Subject<Exception> _errorSubject;
    private readonly BehaviorSubject<bool> _isHealthy;

    public ReactiveMessageBus(ILogger<ReactiveMessageBus> logger, int capacity = 1000)
    {
        _logger = logger;
        _cancellationTokenSource = new CancellationTokenSource();

        // 创建有界Channel，支持高并发
        var options = new BoundedChannelOptions(capacity)
        {
            FullMode = BoundedChannelFullMode.Wait,
            SingleReader = false,
            SingleWriter = false,
            AllowSynchronousContinuations = false
        };

        _channel = Channel.CreateBounded<Message>(options);
        _writer = _channel.Writer;
        _reader = _channel.Reader;

        _messageSubject = new Subject<Message>();
        _errorSubject = new Subject<Exception>();
        _isHealthy = new BehaviorSubject<bool>(true);

        // 按类型过滤消息流
        _dataMessages = _messageSubject.OfType<DataMessage>();
        _eventMessages = _messageSubject.OfType<EventMessage>();
        _commandMessages = _messageSubject.OfType<CommandMessage>();
        _responseMessages = _messageSubject.OfType<ResponseMessage>();

        // 启动消息处理任务
        _processingTask = ProcessMessagesAsync(_cancellationTokenSource.Token);

        // 设置错误处理
        SetupErrorHandling();
    }

    /// <summary>
    /// 获取数据消息流
    /// </summary>
    public IObservable<DataMessage> DataMessages => _dataMessages;

    /// <summary>
    /// 获取事件消息流
    /// </summary>
    public IObservable<EventMessage> EventMessages => _eventMessages;

    /// <summary>
    /// 获取命令消息流
    /// </summary>
    public IObservable<CommandMessage> CommandMessages => _commandMessages;

    /// <summary>
    /// 获取响应消息流
    /// </summary>
    public IObservable<ResponseMessage> ResponseMessages => _responseMessages;

    /// <summary>
    /// 获取错误流
    /// </summary>
    public IObservable<Exception> Errors => _errorSubject.AsObservable();

    /// <summary>
    /// 获取健康状态流
    /// </summary>
    public IObservable<bool> HealthStatus => _isHealthy.AsObservable();

    /// <summary>
    /// 发布消息（异步）
    /// </summary>
    public async ValueTask<bool> PublishAsync(Message message, CancellationToken cancellationToken = default)
    {
        try
        {
            await _writer.WriteAsync(message, cancellationToken);
            _logger.LogDebug("Message published: {MessageId} of type {MessageType}", 
                message.Id, message.GetType().Name);
            return true;
        }
        catch (Exception ex)
        {
            _logger.LogError(ex, "Failed to publish message: {MessageId}", message.Id);
            _errorSubject.OnNext(ex);
            return false;
        }
    }

    /// <summary>
    /// 发布消息（同步，非阻塞）
    /// </summary>
    public bool TryPublish(Message message)
    {
        try
        {
            if (_writer.TryWrite(message))
            {
                _logger.LogDebug("Message published synchronously: {MessageId}", message.Id);
                return true;
            }
            _logger.LogWarning("Channel full, message dropped: {MessageId}", message.Id);
            return false;
        }
        catch (Exception ex)
        {
            _logger.LogError(ex, "Failed to publish message synchronously: {MessageId}", message.Id);
            _errorSubject.OnNext(ex);
            return false;
        }
    }

    /// <summary>
    /// 获取特定源的消息流
    /// </summary>
    public IObservable<T> GetMessagesFromSource<T>(string source) where T : Message
    {
        return _messageSubject
            .OfType<T>()
            .Where(m => m.Source == source);
    }

    /// <summary>
    /// 获取按优先级排序的数据消息流
    /// </summary>
    public IObservable<DataMessage> GetPriorityDataMessages()
    {
        return _dataMessages
            .Buffer(TimeSpan.FromMilliseconds(100)) // 100ms窗口缓冲
            .Where(buffer => buffer.Count > 0)
            .SelectMany(buffer => buffer.OrderByDescending(m => m.Priority));
    }

    /// <summary>
    /// 获取消息统计信息
    /// </summary>
    public IObservable<MessageStats> GetMessageStats(TimeSpan window)
    {
        return _messageSubject
            .Window(window)
            .SelectMany(w => w.ToList()
                .Select(messages => 
                {
                    var stats = messages.GroupBy(m => m.GetType().Name)
                        .ToDictionary(g => g.Key, g => g.Count());
                    return new MessageStats(DateTime.UtcNow, stats);
                }));
    }

    /// <summary>
    /// 实现请求-响应模式
    /// </summary>
    public IObservable<ResponseMessage> SendCommand(CommandMessage command, TimeSpan timeout)
    {
        var correlationId = command.Id;
        
        return _responseMessages
            .Where(r => r.CorrelationId == correlationId)
            .Take(1)
            .Timeout(timeout)
            .Do(_ => { }, ex =>
            {
                if (ex is TimeoutException)
                    _logger.LogWarning("Command timeout: {CommandId}", correlationId);
            });
    }

    private async Task ProcessMessagesAsync(CancellationToken cancellationToken)
    {
        try
        {
            await foreach (var message in _reader.ReadAllAsync(cancellationToken))
            {
                try
                {
                    _messageSubject.OnNext(message);
                    _logger.LogTrace("Message processed: {MessageId}", message.Id);
                }
                catch (Exception ex)
                {
                    _logger.LogError(ex, "Error processing message: {MessageId}", message.Id);
                    _errorSubject.OnNext(ex);
                }
            }
        }
        catch (OperationCanceledException)
        {
            _logger.LogInformation("Message processing cancelled");
        }
        catch (Exception ex)
        {
            _logger.LogError(ex, "Fatal error in message processing");
            _errorSubject.OnNext(ex);
            _isHealthy.OnNext(false);
        }
    }

    private void SetupErrorHandling()
    {
        // 监控错误频率，自动调整健康状态
        _errorSubject
            .Window(TimeSpan.FromMinutes(1))
            .SelectMany(w => w.Count())
            .Subscribe(errorCount =>
            {
                var isHealthy = errorCount < 10; // 每分钟少于10个错误认为是健康的
                _isHealthy.OnNext(isHealthy);
                
                if (!isHealthy)
                    _logger.LogWarning("High error rate detected: {ErrorCount} errors per minute", errorCount);
            });
    }

    public async ValueTask DisposeAsync()
    {
        _writer.TryComplete();
        _cancellationTokenSource.Cancel();
        
        try
        {
            await _processingTask;
        }
        catch (OperationCanceledException)
        {
            // Expected when cancelling
        }

        _messageSubject.Dispose();
        _errorSubject.Dispose();
        _isHealthy.Dispose();
        _cancellationTokenSource.Dispose();
    }
}

/// <summary>
/// 消息统计信息
/// </summary>
public record MessageStats(DateTime Timestamp, Dictionary<string, int> MessageCounts);